package com.kancy.framework.job.starter.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;

import java.util.List;

/**
 * Created by kancy on 2019/07/23.
 * Dataflow类型用于处理数据流，需实现DataflowJob接口。该接口提供2个方法可供覆盖，分别用于抓取(fetchData)和处理(processData)数据。
 *
 * 可通过DataflowJobConfiguration配置是否流式处理。
 * 流式处理数据只有fetchData方法的返回值为null或集合长度为空时，作业才停止抓取，否则作业将一直运行下去；
 * 非流式处理数据则只会在每次作业执行过程中执行一次fetchData方法和processData方法，随即完成本次作业。
 */
public abstract class AbstractDataFlowJob<T> implements DataflowJob<T> {

    private ShardingContextHolder shardingContextHolder = new ShardingContextHolder();

    /**
     * 获取待处理数据.
     *
     * @param shardingContext 分片上下文
     * @return 待处理的数据集合
     */
    @Override
    public final List<T> fetchData(ShardingContext shardingContext) {
        try {
            shardingContextHolder.set(shardingContext);
            /**
             * 实际开发中，有了任务总片数和当前分片项，就可以对任务进行分片执行了
             * SELECT * FROM table WHERE status = 0 AND MOD(id, shardingTotalCount) = shardingItem Limit 0, 30
             */
            return fetchData();
        } finally {
            shardingContextHolder.remove();
        }
    }

    /**
     * 获取待处理数据.
     * @return 待处理的数据集合
     */
    protected abstract List<T> fetchData();

    /**
     * 处理数据.
     *
     * @param shardingContext 分片上下文
     * @param data            待处理数据集合
     */
    @Override
    public final void processData(ShardingContext shardingContext, List<T> data) {
        try {
            shardingContextHolder.set(shardingContext);
            processData(data);
        } finally {
            shardingContextHolder.remove();
        }
    }

    /**
     * 处理数据.
     *
     * @param data            待处理数据集合
     */
    protected abstract void processData(List<T> data);
}
